package com.exchange.plugin.kafka;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;


public class TestKafkaConsumer {
	public static void main(String[] args) throws InterruptedException, ExecutionException {
		Properties props = new Properties();
    	props.put("bootstrap.servers", "127.0.0.1:9092");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.LongDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("group.id", "matchTrade");
        KafkaConsumer<Long, String> consumer = new KafkaConsumer<Long, String>(props);
        consumer.subscribe(Collections.singletonList("matchTrade"));//订阅TOPIC
        
        try {
		    while (true) {
		        // 100 是超时时间（ms），在该时间内 poll 会等待服务器返回数据
		    	@SuppressWarnings("deprecation")
				ConsumerRecords<Long, String> records = consumer.poll(Duration.ofMillis(1000)); 
		        // poll 返回一个记录列表。
		        // 每条记录都包含了记录所属主题的信息、记录所在分区的信息、记录在分区里的偏移量，以及记录的键值对。(Trade部分)
		    	//country={"isBuy":true,"uid":455951945836789761,"number":1,"dealWay":"maker","coinTeam":"XBIT-USDT","totalPrice":10,"price":10,"oid":455951945836789760}
		        for (ConsumerRecord<Long, String> record : records) {
			        //System.out.println("topic="+record.topic()+", partition="+record.partition()+", offset="+record.offset()+", customer="+record.key()+", country="+record.value());
//		        	System.out.println("获取撮合信息");
			    }
		        /*consumer.commitSync();//同步*/
		    } 
		} finally {
		    // 关闭消费者,网络连接和 socket 也会随之关闭，并立即触发一次再均衡
			consumer.close();
		}
        
    }
}
